use std::{fmt, sync::Arc, time::{Duration, Instant}};

use parking_lot::{Condvar, Mutex, MutexGuard};

use crate::{config::Config, conn_utils, event::TimeoutEvent, Builder, CheckinEvent, CheckoutEvent, Error, Extensions, ManageConnection, PooledConnection, State};



pub(crate) struct Conn<C> {
    pub(crate) conn: C,
    pub(crate) extensions: Extensions,
    pub(crate) birth: Instant,
    pub(crate) id: u64,
}

pub(crate) struct IdleConn<C> {
    pub(crate) conn: Conn<C>,
    pub(crate) idle_start: Instant,
}

pub(crate) struct PoolInternals<C> {
    pub(crate) conns: Vec<IdleConn<C>>,
    pub(crate) num_conns: u32,
    pub(crate) pending_conns: u32,
    pub(crate) last_error: Option<String>,
}

pub(crate) struct SharedPool<M>
where
    M: ManageConnection,
{
    pub(crate) config: Config<M::Connection, M::Error>,
    pub(crate) manager: M,
    pub(crate) internals: Mutex<PoolInternals<M::Connection>>,
    pub(crate) cond: Condvar,
}
/// A generic connection pool.
pub struct Pool<M>(Arc<SharedPool<M>>)
where
    M: ManageConnection;

/// Returns a new `Pool` referencing the same state as `self`.
impl<M> Clone for Pool<M>
where
    M: ManageConnection,
{
    fn clone(&self) -> Pool<M> {
        Pool(self.0.clone())
    }
}

impl<M> fmt::Debug for Pool<M>
where
    M: ManageConnection + fmt::Debug,
{
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Pool")
            .field("state", &self.state())
            .field("config", &self.0.config)
            .field("manager", &self.0.manager)
            .finish()
    }
}

impl<M> Pool<M>
where
    M: ManageConnection,
{
    /// Creates a new connection pool with a default configuration.
    pub fn new(manager: M) -> Result<Pool<M>, Error> {
        Pool::builder().build(manager)
    }

    /// Returns a builder type to configure a new pool.
    pub fn builder() -> Builder<M> {
        Builder::new()
    }

    // for testing
    pub(crate) fn new_inner(
        config: Config<M::Connection, M::Error>,
        manager: M,
        reaper_rate: Duration,
    ) -> Pool<M> {
        let internals = PoolInternals {
            conns: Vec::with_capacity(config.max_size as usize),
            num_conns: 0,
            pending_conns: 0,
            last_error: None,
        };

        let shared = Arc::new(SharedPool {
            config,
            manager,
            internals: Mutex::new(internals),
            cond: Condvar::new(),
        });

        conn_utils::establish_idle_connections(&shared, &mut shared.internals.lock());

        if shared.config.max_lifetime.is_some() || shared.config.idle_timeout.is_some() {
            let s = Arc::downgrade(&shared);
            shared
                .config
                .thread_pool
                .execute_at_fixed_rate(reaper_rate, reaper_rate, move || conn_utils::reap_connections(&s));
        }

        Pool(shared)
    }

    pub(crate)  fn wait_for_initialization(&self) -> Result<(), Error> {
        let end = Instant::now() + self.0.config.connection_timeout;
        let mut internals = self.0.internals.lock();

        let initial_size = self.0.config.min_idle.unwrap_or(self.0.config.max_size);

        while internals.num_conns != initial_size {
            if self.0.cond.wait_until(&mut internals, end).timed_out() {
                return Err(Error(internals.last_error.take()));
            }
        }

        Ok(())
    }

    /// Retrieves a connection from the pool.
    ///
    /// Waits for at most the configured connection timeout before returning an
    /// error.
    pub fn get(&self) -> Result<PooledConnection<M>, Error> {
        self.get_timeout(self.0.config.connection_timeout)
    }

    /// Retrieves a connection from the pool, waiting for at most `timeout`
    ///
    /// The given timeout will be used instead of the configured connection
    /// timeout.
    pub fn get_timeout(&self, timeout: Duration) -> Result<PooledConnection<M>, Error> {
        let start = Instant::now();
        let end = start + timeout;
        let mut internals = self.0.internals.lock();

        loop {
            match self.try_get_inner(internals) {
                Ok(conn) => {
                    let event = CheckoutEvent {
                        id: conn.conn.as_ref().unwrap().id,
                        duration: start.elapsed(),
                    };
                    self.0.config.event_handler.handle_checkout(event);
                    return Ok(conn);
                }
                Err(i) => internals = i,
            }

            conn_utils::add_connection(&self.0, &mut internals);

            if self.0.cond.wait_until(&mut internals, end).timed_out() {
                let event = TimeoutEvent { timeout };
                self.0.config.event_handler.handle_timeout(event);

                return Err(Error(internals.last_error.take()));
            }
        }
    }

    /// Attempts to retrieve a connection from the pool if there is one
    /// available.
    ///
    /// Returns `None` if there are no idle connections available in the pool.
    /// This method will not block waiting to establish a new connection.
    pub fn try_get(&self) -> Option<PooledConnection<M>> {
        self.try_get_inner(self.0.internals.lock()).ok()
    }

    fn try_get_inner<'a>(
        &'a self,
        mut internals: MutexGuard<'a, PoolInternals<M::Connection>>,
    ) -> Result<PooledConnection<M>, MutexGuard<'a, PoolInternals<M::Connection>>> {
        loop {
            if let Some(mut conn) = internals.conns.pop() {
                conn_utils::establish_idle_connections(&self.0, &mut internals);
                drop(internals);

                if self.0.config.test_on_check_out {
                    if let Err(e) = self.0.manager.is_valid(&mut conn.conn.conn) {
                        let msg = e.to_string();
                        self.0.config.error_handler.handle_error(e);
                        // FIXME we shouldn't have to lock, unlock, and relock here
                        internals = self.0.internals.lock();
                        internals.last_error = Some(msg);
                        conn_utils::drop_conns(&self.0, internals, vec![conn.conn]);
                        internals = self.0.internals.lock();
                        continue;
                    }
                }

                return Ok(PooledConnection {
                    pool: self.clone(),
                    checkout: Instant::now(),
                    conn: Some(conn.conn),
                });
            } else {
                return Err(internals);
            }
        }
    }

    pub(crate) fn put_back(&self, checkout: Instant, mut conn: Conn<M::Connection>) {
        let event = CheckinEvent {
            id: conn.id,
            duration: checkout.elapsed(),
        };
        self.0.config.event_handler.handle_checkin(event);

        // This is specified to be fast, but call it before locking anyways
        let broken = self.0.manager.has_broken(&mut conn.conn);

        let mut internals = self.0.internals.lock();
        if broken {
            conn_utils::drop_conns(&self.0, internals, vec![conn]);
        } else {
            let conn = IdleConn {
                conn,
                idle_start: Instant::now(),
            };
            internals.conns.push(conn);
            self.0.cond.notify_one();
        }
    }

    /// Returns information about the current state of the pool.
    pub fn state(&self) -> State {
        let internals = self.0.internals.lock();
        State {
            connections: internals.num_conns,
            idle_connections: internals.conns.len() as u32,
        }
    }

    /// Returns the configured maximum pool size.
    pub fn max_size(&self) -> u32 {
        self.0.config.max_size
    }

    /// Returns the configured minimum idle connection count.
    pub fn min_idle(&self) -> Option<u32> {
        self.0.config.min_idle
    }

    /// Returns if the pool is configured to test connections on check out.
    pub fn test_on_check_out(&self) -> bool {
        self.0.config.test_on_check_out
    }

    /// Returns the configured maximum connection lifetime.
    pub fn max_lifetime(&self) -> Option<Duration> {
        self.0.config.max_lifetime
    }

    /// Returns the configured idle connection timeout.
    pub fn idle_timeout(&self) -> Option<Duration> {
        self.0.config.idle_timeout
    }

    /// Returns the configured connection timeout.
    pub fn connection_timeout(&self) -> Duration {
        self.0.config.connection_timeout
    }
}
